package com.xunk.equipment.websocket;

import cn.hutool.core.util.StrUtil;
import com.xunk.common.core.text.Convert;
import com.xunk.common.core.utils.SpringUtils;
import com.xunk.common.core.utils.serialization.JsonSerializer;
import com.xunk.equipment.config.ZhijiaConfig;
import com.xunk.equipment.service.zhijia.constant.ZhijiaConstant;
import com.xunk.equipment.service.zhijia.enums.ZhijiaTopicEnum;
import com.xunk.equipment.service.zhijia.service.ZhijiaService;
import com.xunk.equipment.service.zhijia.util.ZhijiaUtil;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.tomcat.websocket.WsWebSocketContainer;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import java.io.IOException;
import java.net.URI;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import static java.util.concurrent.Executors.newFixedThreadPool;

/**
 * @ProjectName: HIM
 * @Package: com.xunk.equipment.websocket
 * @ClassName: ZhijiaWebSocketClient
 * @Author: luanzhiwei
 * @Description: 智家websocket客户端
 * @Date: 2021/4/23 15:46
 * @Version: 1.0
 * 详情请看 https://haier-iot.github.io/doc/#/zh-cn/DevicesManage/datasubscription?id=%E5%8F%96%E6%B6%88%E8%AE%A2%E9%98%85%E6%8E%A5%E5%8F%A3
 */
@ClientEndpoint(encoders = {TextEncoder.class})
@Slf4j
@NoArgsConstructor
@Component
public class ZhijiaWebSocketClient implements ApplicationRunner {

    //该类中不要使用 @Autowired 注入bean，无法注入，会报空指针异常

    // 最新收消息时间,心跳发消息时做时间间隔计算
    public static AtomicLong lastReceiveTime = null;
    // 是否继续心跳标志位
    public static AtomicBoolean isHeartBeatContinue = null;
    private DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
    // 创建容器对象（建立连接时要用）
    private WebSocketContainer container = ContainerProvider.getWebSocketContainer();
    // 一个连接对应一个 session
    private Session session = null;
    // 心跳线程池
    private ExecutorService executorService = null;
    // 关闭心跳线程要用到
    private Future<?> future = null;

    /**
     * 连接到服务端
     *
     * @param uri
     * @throws DeploymentException
     * @throws IOException
     */
    public void connect(String uri) throws DeploymentException, IOException {
        session = container.connectToServer(ZhijiaWebSocketClient.class, URI.create(uri));
        if (session.isOpen()) {
            executorService = newFixedThreadPool(1);
            lastReceiveTime = new AtomicLong(System.currentTimeMillis());
            isHeartBeatContinue = new AtomicBoolean(true);
            session.setMaxIdleTimeout(0);// 值为 0 表示会话连接不会因为长时间无数据交互而超时
            session.setMaxTextMessageBufferSize(1024 * 1024); //1Mb
            startHeartBeat();// 启动心跳
        } else {
            container = null;
        }

    }


    /**
     * 当连接服务端成功，回调该方法
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        log.info("连接智家websocket服务端，连接结果：{}", session.isOpen());
    }


    /**
     * 当收到服务端消息，回调该方法
     *
     * @param message
     */
    @OnMessage
    public void onMessage(String message) {
        try {

            long currentTime = System.currentTimeMillis();
            // 更新最新收消息时间
            lastReceiveTime.set(currentTime);
            String currentDateStr = dateFormat.format(currentTime);
            ZhijiaService bean = SpringUtils.getBean(StrUtil.lowerFirst(ZhijiaService.class.getSimpleName()));
            List<String> typeIds = ZhijiaUtil.getTypeIds();
            if (message.contains("-ack")) {
                log.info(currentDateStr + " 收到系统答信息： " + message);
                bean.processAck(message);
            } else {
                Map<String, Object> deserializeMap = JsonSerializer.deserializeMap(message);
                Map body = (Map) deserializeMap.get("body");
                String typeId = Convert.toStr(body.get("typeId"));
                if (!typeIds.contains(typeId)) {
                    log.info("当前推送设备typeid：{}不存在，不消费", typeId);
                    return;
                }
                log.info(currentDateStr + " 收到 message： " + message);
                bean.processMessage(message);
            }
        } catch (Exception e) {
            log.info("onMessage收到智家推送处理异常: {}", e);
        }

    }

    /**
     * 当异常时，回调该方法
     *
     * @param t
     */
    @OnError
    public void onError(Throwable t) {
        t.printStackTrace();
        log.info(t.getMessage());
    }

    /**
     * 当断开连接时调用该方法
     */
    @OnClose
    public void onClose(Session session, CloseReason closeReason) {
        log.info("客户端与智家断开连接");
        stopHeartBeat();
        stopContainer();
        final int code = closeReason.getCloseCode().getCode();
        if (code == 1006 || code == 4500) {
            // 1006：连接异常关闭 4500：通道异常，服务器端主动断开连接
            boolean isSuccess = false;
            while (!isSuccess) {
                try {
                    log.info("Try restarting the connection after 5s...");
                    Thread.sleep(5000);
                    isSuccess = buildConnectionAndSubscribeMsg();
                    if (!isSuccess) {
                        log.info("Try restarting the connection after 2min...");
                        Thread.sleep(120000);
                        isSuccess = buildConnectionAndSubscribeMsg();
                        if (!isSuccess) {
                            log.info("Try restarting the connection after 5min...");
                            Thread.sleep(300000);
                            isSuccess = buildConnectionAndSubscribeMsg();
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * 添加命令
     *
     * @param cmdInfo
     */
    public void addCmdInfo(final String cmdInfo) {
        new Thread(() -> {
            try {
                if (session.isOpen()) {
                    synchronized (session) {
                        session.getBasicRemote().sendObject(cmdInfo);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).start();
    }


    /**
     * 关闭连接
     */
    public void close() {
        try {
            // 关闭心跳
            stopHeartBeat();
            // 关闭 session 连接
            session.close(new CloseReason(CloseReason.CloseCodes.NORMAL_CLOSURE, ""));
            // 关闭容器
            stopContainer();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 如果项目引用的是 Tomcat 的 websocket 实现，则采用如下代码
     * import org.apache.tomcat.websocket.WsWebSocketContainer;
     */
    private void stopContainer() {
        log.info("stopContainer");
        if (container instanceof WsWebSocketContainer) {
            try {
                ((WsWebSocketContainer) container).destroy();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 开启心跳
     */
    private void startHeartBeat() {
        future = executorService.submit(new HeartBeat());
    }

    /**
     * 停止心跳
     */
    private void stopHeartBeat() {
        log.info("stopHeartBeat");
        if (future != null) {
            try {
                isHeartBeatContinue.set(false);
                future.cancel(true);
                executorService.shutdown();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private class HeartBeat implements Runnable {
        // 心跳间隔时间，具体根据实际情况设置,如 nginx 默认 websocket 超时时间是 60s，则
//        可以在连接无数据交互情况持续到快过期前发送一次心跳（ 这里设置 55s 时），
        private long heartBeatInterval = 55000;

        @Override
        public void run() {
            try {
                while (isHeartBeatContinue.get()) {
                    long startTime = System.currentTimeMillis();
                    if (startTime - lastReceiveTime.get() > heartBeatInterval) {
                        String cmdInfo = "{'cmd': 'keepAlive'}";
                        if (session.isOpen()) {
                            synchronized (session) {
                                session.getBasicRemote().sendObject(cmdInfo);
                            }
                            // 更新最新收消息时间
                            lastReceiveTime.set(System.currentTimeMillis());
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 拼接订阅报文信息，不含cmd参数
     *
     * @return
     */
    private Map<String, Object> getSubreq() {
        Map<String, Object> request = new HashMap<>();
        List<String> list = new ArrayList<>();
        for (ZhijiaTopicEnum value : ZhijiaTopicEnum.values()) {
            list.add(value.getCode());
        }
        request.put("topics", list);
        return request;
    }

    /**
     * 创建连接
     *
     * @return
     */
    private boolean buildConnectionAndSubscribeMsg() {
        long currentTimeMillis = System.currentTimeMillis();
        final ZhijiaConfig config = SpringUtils.getBean(StrUtil.lowerFirst(ZhijiaConfig.class.getSimpleName()));
        String uri = String.format("%s?systemId=%s&timestamp=%s&sign=%s", config.getSocketUri(), config.getSystemId(), currentTimeMillis, ZhijiaUtil.signSocket(config.getSystemId(), config.getSystemKey(), Convert.toStr(currentTimeMillis)));
        log.info("智家websocket订阅uri: {}", uri);
        Map<String, Object> subreq = getSubreq();
        subreq.put("cmd", "subscribe");
        String cmdInfo = JsonSerializer.serialize(subreq);
        log.info("智家websocket订阅报文: {}", cmdInfo);
        try {
            // 建立连接
            this.connect(uri);
            // 发送订阅信息
            this.addCmdInfo(cmdInfo);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    /**
     * 取消订阅
     *
     * @return
     */
    public boolean unsubscribe() {
        log.info("取消订阅开始");
        Map<String, Object> subreq = getSubreq();
        subreq.put("cmd", "unsubscribe");
        try {
            this.addCmdInfo(JsonSerializer.serialize(subreq));
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        log.info("取消订阅结束");
        return true;
    }

    /**
     * 查询当前system的订阅关系
     */
    public void querySubscribe() {
        Map<String, Object> cmd = new HashMap<>();
        cmd.put("cmd", "searchSystem");
        this.addCmdInfo(JsonSerializer.serialize(cmd));
    }

    /**
     * 服务启动，建立连接，开始订阅
     * 客户端发起连接请求， 请求时携带对应应用的 systemId 和 systemKey 签名信息进行
     * 鉴权，鉴权成功返回鉴权结果，鉴权失败，则返回错误信息并断开连接。
     * 默认所有设备订阅全部topic
     *
     * @param args
     * @throws Exception
     */
    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("服务启动，与智家websocketserver建立连接并开始订阅");
        boolean flag = buildConnectionAndSubscribeMsg();
        log.info("建立连接+订阅服务结果：{}", flag);
    }
}
